package com.metis.document.parse.dialog.config;

import static com.metis.document.parse.dialog.service.message_listener.ParseTaskMessageListener.STREAM_GROUP;
import static com.metis.document.parse.dialog.service.message_listener.ParseTaskMessageListener.STREAM_KEY;

import java.time.Duration;
import java.util.concurrent.ThreadPoolExecutor;

import javax.annotation.Resource;

import com.metis.document.parse.dialog.service.message_listener.DoNothingMessageListener;
import com.metis.document.parse.dialog.service.message_listener.ParseTaskMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;


@Slf4j
@Configuration
public class RedisConfig {

    @Autowired
    private StreamListener<String, ObjectRecord<String, String>> doNothingMessageListener;
    @Autowired
    private StreamListener<String, ObjectRecord<String, String>> parseTaskMessageListener;
    @Value("${threadPool.heightTaskExecutor.threadCount:10}")
    private int heightTaskExecutorThreadCount;
    @Value("${threadPool.lowTaskExecutor.threadCount:10}")
    private int lowTaskExecutorThreadCount;
    @Value("${threadPool.parseTaskExecutor.threadCount:10}")
    private int parseTaskExecutorThreadCount;

    @Bean({"redisStringTemplate"})
    public RedisTemplate<String, String> redisStringTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringRedisSerializer);
        redisTemplate.setValueSerializer(stringRedisSerializer);
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        redisTemplate.setHashValueSerializer(stringRedisSerializer);
        return redisTemplate;
    }
    @Bean("parseTaskExecutor")
    public ThreadPoolTaskExecutor parseTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(parseTaskExecutorThreadCount);
        executor.setMaxPoolSize(parseTaskExecutorThreadCount);
        executor.setQueueCapacity(0);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("知识库解析任务-");
        executor.setWaitForTasksToCompleteOnShutdown(false);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setAwaitTerminationSeconds(60);
        return executor;
    }
    @Bean("heightTaskExecutor")
    public ThreadPoolTaskExecutor heightTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(heightTaskExecutorThreadCount);
        executor.setMaxPoolSize(heightTaskExecutorThreadCount);
        executor.setQueueCapacity(0);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("大文件解析任务-");
        executor.setWaitForTasksToCompleteOnShutdown(false);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setAwaitTerminationSeconds(60);
        return executor;
    }
    @Bean("lowTaskExecutor")
    public ThreadPoolTaskExecutor lowTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(lowTaskExecutorThreadCount);
        executor.setMaxPoolSize(lowTaskExecutorThreadCount);
        executor.setQueueCapacity(0);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("小文件解析任务-");
        executor.setWaitForTasksToCompleteOnShutdown(false);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setAwaitTerminationSeconds(60);
        return executor;
    }

    @Bean(initMethod = "start")
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> parseTaskContainer(RedisTemplate<String, String> redisStringTemplate ,RedisConnectionFactory connectionFactory) {
        if (Boolean.FALSE.equals(redisStringTemplate.hasKey(STREAM_KEY))
                || redisStringTemplate.opsForStream().groups(STREAM_KEY).stream().noneMatch(x -> x.groupName().equals(STREAM_GROUP))) {
            redisStringTemplate.opsForStream().createGroup(STREAM_KEY, STREAM_GROUP);
        }
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(1000)
                        // 可以理解为 Stream Key 的序列化方式
                        .keySerializer(RedisSerializer.string())
                        // 可以理解为 Stream 后方的字段的 key 的序列化方式
                        .hashKeySerializer(RedisSerializer.string())
                        // 可以理解为 Stream 后方的字段的 value 的序列化方式
                        .hashValueSerializer(RedisSerializer.string())
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                        .pollTimeout(Duration.ofSeconds(1))
                        // ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比如：将Book对象转换成map
                        .objectMapper(new ObjectHashMapper())
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(t -> log.error("StreamMessageListenerContainer error", t))
                        // 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型是这个地方指定的类型
                        .targetType(String.class)
                        .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer
                        .create(connectionFactory, options);
        StreamOffset<String> last = StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed());
        StreamOffset<String> start = StreamOffset.fromStart(STREAM_KEY);
        //设置重新消费未ack的消息,先获取消息，再ack
        streamMessageListenerContainer
                .receive(Consumer.from(STREAM_GROUP, STREAM_GROUP),
                        start, parseTaskMessageListener);
        streamMessageListenerContainer
                .receive(Consumer.from(STREAM_GROUP, STREAM_GROUP),
                        last, doNothingMessageListener);
        return streamMessageListenerContainer;
    }

}
